package cn.aoyu.test.tools.kafka;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 接收数据
 * 接收到: message: 10
 * 接收到: message: 11
 * 接收到: message: 12
 * 接收到: message: 13
 * 接收到: message: 14
 *
 * @author xww
 */
public class KafkaConsumer extends Thread {

	private static final Logger log = LoggerFactory.getLogger(KafkaConsumer.class);

	private String topic;

	public KafkaConsumer(String topic) {
		super();
		this.topic = topic;
	}


	@Override
	public void run() {
		ConsumerConnector consumer = createConsumer();
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, 1); // 一次从主题中获取一个数据
		Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
		KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据
		ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
		while (iterator.hasNext()) {
			String message = new String(iterator.next().message());
			System.out.println("接收到: " + message);
		}
	}

	private ConsumerConnector createConsumer() {
		Properties properties = new Properties();
		properties.put("zookeeper.connect", "192.168.10.58:2181");//声明zk
		properties.put("group.id", "group1");// 必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据
		return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
	}


	public static void main(String[] args) {
		new KafkaConsumer("test").start();// 使用kafka集群中创建好的主题 test
	}

	/**
	 * 将byte数组转化为Object对象
	 *
	 * @return
	 */
	private Object toObject(byte[] bytes) {
		Object object = null;
		try {
			ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);// 创建ByteArrayInputStream对象
			ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);// 创建ObjectInputStream对象
			object = objectInputStream.readObject();// 从objectInputStream流中读取一个对象
			byteArrayInputStream.close();// 关闭输入流
			objectInputStream.close();// 关闭输入流
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
		return object;// 返回对象
	}

}
